package com.flying.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

public class DataProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.56.88:9093,192.168.56.88:9095,192.168.56.88:9097");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.KafkaAvroSerializer");
        props.put("schema.registry.url", "http://192.168.56.88:8085");
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.flying.kafka.interceptor.DataProducerMetricsInterceptor");

        Producer<String, Cat> producer = new KafkaProducer<>(props);

        Cat cat = Cat.newBuilder().setBreed(Breed.ABYSSINIAN).build();
        System.out.println(cat);
        ProducerRecord<String, Cat> record = new ProducerRecord<>("helloworld", cat.getBreed().name(), cat);

        producer.send(record);
        producer.close();
    }
}
